package com.gph.saviorframework.mq.api.impl.rocketmq;

import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.gph.saviorframework.mq.api.Message;
import com.gph.saviorframework.mq.api.PropertyKeyConst;
import com.gph.saviorframework.mq.api.SendResult;
import com.gph.saviorframework.mq.api.exception.MQClientException;
import com.gph.saviorframework.mq.api.impl.util.MQUtil;
import com.gph.saviorframework.mq.api.order.OrderProducer;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by root on 16-7-11.
 */
public class OrderProducerImpl extends MQClientAbstract implements OrderProducer {

    private final DefaultMQProducer defaultMQProducer;
    private final AtomicBoolean started = new AtomicBoolean(false);

    public OrderProducerImpl(final Properties properties){

        super(properties);
        this.defaultMQProducer = new DefaultMQProducer(new ClientRPCHook(sessionCredentials));

        String producerGroup = properties.getProperty(PropertyKeyConst.ProducerId, "__GPH_PRODUCER_DEFAULT_GROUP");
        this.defaultMQProducer.setProducerGroup(producerGroup);

        boolean isVipChannelEnabled = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.isVipChannelEnabled, "false"));
        this.defaultMQProducer.setVipChannelEnabled(isVipChannelEnabled);

        String sendMsgTimeoutMillis = properties.getProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
        this.defaultMQProducer.setSendMsgTimeout(Integer.parseInt(sendMsgTimeoutMillis));

        this.defaultMQProducer.setInstanceName(this.buildIntanceName());
        this.defaultMQProducer.setNamesrvAddr(this.getNameServerAddr());
    }

    /**
     * 启动服务
     */
    public void start() {
        try {
            if (started.compareAndSet(false, true)) {
                this.defaultMQProducer.start();
            }
        }
        catch (Exception e) {
            throw new MQClientException(e.getMessage());
        }
    }

    /**
     * 关闭服务
     */
    public void shutdown() {
        if (started.compareAndSet(true, false)) {
            this.defaultMQProducer.shutdown();
        }
    }

    /**
     * 发送顺序消息
     *
     * @param message     消息
     * @param shardingKey 顺序消息选择因子
     * @return 发送结果，含消息Id
     */
    public SendResult send(Message message, String shardingKey) {
        if (UtilAll.isBlank(shardingKey)) {
            throw new MQClientException("\'shardingKey\' is blank.");
        }

        this.checkProducerServiceState(this.defaultMQProducer.getDefaultMQProducerImpl());
        final com.alibaba.rocketmq.common.message.Message msgRMQ = MQUtil.msgConvert(message);
        try {
            com.alibaba.rocketmq.client.producer.SendResult sendResultRMQ =
                    this.defaultMQProducer.send(msgRMQ, new com.alibaba.rocketmq.client.producer.MessageQueueSelector() {
                        public MessageQueue select(List<MessageQueue> mqs, com.alibaba.rocketmq.common.message.Message msg,
                                                   Object shardingKey) {
                            int select = Math.abs(shardingKey.hashCode());
                            if (select < 0) {
                                select = 0;
                            }
                            return mqs.get(select % mqs.size());
                        }
                    }, shardingKey);
            SendResult sendResult = new SendResult();
            sendResult.setMessageId(sendResultRMQ.getMsgId());
            return sendResult;
        }
        catch (Exception e) {
            throw new MQClientException("defaultMQProducer send order exception", e);
        }
    }

    public boolean isStarted() {
        return started.get();
    }

    public boolean isClosed() {
        return !isStarted();
    }
}
